package com.zhucan.order.application.command.impl;

import com.zhucan.common.commodity.CommodityProvider;
import com.zhucan.common.commodity.dto.CommodityDTO;
import com.zhucan.common.order.event.OrderCreateEvent;
import com.zhucan.order.application.command.OrderCommandService;
import com.zhucan.order.application.command.cmd.OrderCreateCommand;
import com.zhucan.order.infrastructure.db.entity.OrderPO;
import com.zhucan.order.infrastructure.db.repository.OrderRepository;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.dubbo.config.annotation.Reference;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFutureCallback;

/**
 * @author zhuCan
 * @description
 * @since 2022-07-06 16:58
 **/
@Slf4j
@Service
@RequiredArgsConstructor
public class OrderCommandServiceImpl implements OrderCommandService {

    private final OrderRepository orderRepository;

    @Reference
    private CommodityProvider commodityProvider;

    private final KafkaTemplate<String, Object> kafkaTemplate;


    @Override
    public void create(OrderCreateCommand command) {
        CommodityDTO commodity = commodityProvider.info(command.getCommodityId());

        // 订单保存
        OrderPO order = orderRepository.save(OrderPO.builder()
                .commodityId(commodity.getId())
                .amount(commodity.getPrice())
                .commodityName(commodity.getName())
                .creator(command.getCreator())
                .build());

        // 发布下单事件
        kafkaTemplate.send("order_event", new OrderCreateEvent(order.getId(), order.getCommodityId()))
                .addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
                    @Override
                    public void onFailure(Throwable ex) {
                        log.error("发送失败!", ex);
                    }

                    @Override
                    public void onSuccess(SendResult<String, Object> result) {
                        log.info("发送成功:{}", result.getProducerRecord().value());
                    }
                });

    }
}
